Clickx 115/Clickx 115.iso/software/tools/windows/tails-i386-0.16.iso/live/filesystem.squashfs/usr/share/pyshared/gnutls/connection.py

Python Source | 2008-04-10 | 19.3 KB | 501 lines

# Copyright (C) 2007-2008 AG Projects. See LICENSE for details.
#

"""GNUTLS connection support"""

__all__ = ['X509Credentials', 'ClientSession', 'ServerSession', 'ServerSessionFactory']

from time import time
from socket import SHUT_RDWR as SOCKET_SHUT_RDWR

from _ctypes import PyObj_FromPtr
from ctypes import *

from gnutls.validators import *
from gnutls.constants import *
from gnutls.crypto import *
from gnutls.errors import *

from gnutls.library.constants import GNUTLS_SERVER, GNUTLS_CLIENT, GNUTLS_CRT_X509
from gnutls.library.constants import GNUTLS_CERT_INVALID, GNUTLS_CERT_REVOKED, GNUTLS_CERT_INSECURE_ALGORITHM
from gnutls.library.constants import GNUTLS_CERT_SIGNER_NOT_FOUND, GNUTLS_CERT_SIGNER_NOT_CA
from gnutls.library.constants import GNUTLS_AL_FATAL, GNUTLS_A_BAD_CERTIFICATE
from gnutls.library.constants import GNUTLS_A_UNKNOWN_CA, GNUTLS_A_INSUFFICIENT_SECURITY
from gnutls.library.constants import GNUTLS_A_CERTIFICATE_EXPIRED, GNUTLS_A_CERTIFICATE_REVOKED
from gnutls.library.constants import GNUTLS_NAME_DNS
from gnutls.library.types     import gnutls_certificate_credentials_t, gnutls_session_t, gnutls_x509_crt_t
from gnutls.library.types     import gnutls_certificate_server_retrieve_function
from gnutls.library.functions import *


@gnutls_certificate_server_retrieve_function
def _retrieve_server_certificate(c_session, retr_st):
    session = PyObj_FromPtr(gnutls_session_get_ptr(c_session))
    identity = session.credentials.select_server_identity(session)
    retr_st.contents.type = GNUTLS_CRT_X509
    retr_st.contents.deinit_all = 0
    if identity is None:
        retr_st.contents.ncerts = 0
    else:
        retr_st.contents.ncerts = 1
        retr_st.contents.cert.x509.contents = identity.cert._c_object
        retr_st.contents.key.x509 = identity.key._c_object
    return 0


class _ServerNameIdentities(dict):
    """Used internally by X509Credentials to map server names to X509 identities for the server name extension"""
    def __init__(self, identities):
        dict.__init__(self)
        for identity in identities:
            self.add(identity)
    def add(self, identity):
        for name in identity.cert.alternative_names.dns:
            self[name.lower()] = identity
        for ip in identity.cert.alternative_names.ip:
            self[ip] = identity
        subject = identity.cert.subject
        if subject.CN is not None:
            self[subject.CN.lower()] = identity
    def get(self, server_name, default=None):
        server_name = server_name.lower()
        if server_name in self:
            return self[server_name]
        for name in (n for n in self if n.startswith('*.')):
            suffix = name[1:]
            if server_name.endswith(suffix) and '.' not in server_name[:-len(suffix)]:
                return self[name]
        return default


class X509Credentials(object):
    DH_BITS  = 1024
    RSA_BITS = 1024

    dh_params  = None
    rsa_params = None

    def __new__(cls, *args, **kwargs):
        c_object = gnutls_certificate_credentials_t()
        gnutls_certificate_allocate_credentials(byref(c_object))
        instance = object.__new__(cls)
        instance.__deinit = gnutls_certificate_free_credentials
        instance._c_object = c_object
        return instance

    @method_args((X509Certificate, none), (X509PrivateKey, none), list_of(X509Certificate), list_of(X509CRL), list_of(X509Identity))
    def __init__(self, cert=None, key=None, trusted=[], crl_list=[], identities=[]):
        """Credentials contain a X509 certificate, a private key, a list of trusted CAs and a list of CRLs (all optional).
        An optional list of additional X509 identities can be specified for applications that need more that one identity"""
        if cert and key:
            gnutls_certificate_set_x509_key(self._c_object, byref(cert._c_object), 1, key._c_object)
        elif (cert, key) != (None, None):
            raise ValueError("Specify neither or both the certificate and private key")
        gnutls_certificate_server_set_retrieve_function(self._c_object, _retrieve_server_certificate)
        self._max_depth = 5
        self._max_bits  = 8200
        self._type = CRED_CERTIFICATE
        self._cert = cert
        self._key = key
        self._identities = tuple(identities)
        self._trusted = ()
        self.add_trusted(trusted)
        self.crl_list = crl_list
        self.server_name_identities = _ServerNameIdentities(identities)
        if cert and key:
            self.server_name_identities.add(X509Identity(cert, key))
        self.session_params = SessionParams(self._type)

    def __del__(self):
        self.__deinit(self._c_object)

    # Methods to alter the credentials at runtime

    @method_args(list_of(X509Certificate))
    def add_trusted(self, trusted):
        size = len(trusted)
        if size > 0:
            ca_list = (gnutls_x509_crt_t * size)(*[cert._c_object for cert in trusted])
            gnutls_certificate_set_x509_trust(self._c_object, cast(byref(ca_list), POINTER(gnutls_x509_crt_t)), size)
            self._trusted = self._trusted + tuple(trusted)

    def generate_dh_params(self, bits=DH_BITS):
        reference = self.dh_params ## keep a reference to preserve it until replaced
        X509Credentials.dh_params  = DHParams(bits)
        del reference

    def generate_rsa_params(self, bits=RSA_BITS):
        reference = self.rsa_params ## keep a reference to preserve it until replaced
        X509Credentials.rsa_params = RSAParams(bits)
        del reference

    # Properties

    @property
    def cert(self):
        return self._cert

    @property
    def key(self):
        return self._key

    @property
    def identities(self):
        return self._identities

    @property
    def trusted(self):
        return self._trusted

    def _get_crl_list(self):
        return self._crl_list
    @method_args(list_of(X509CRL))
    def _set_crl_list(self, crl_list):
        self._crl_list = tuple(crl_list)
    crl_list = property(_get_crl_list, _set_crl_list)
    del _get_crl_list, _set_crl_list

    def _get_max_verify_length(self):
        return self._max_depth
    @method_args(int)
    def _set_max_verify_length(self, max_depth):
        gnutls_certificate_set_verify_limits(self._c_object, self._max_bits, max_depth)
        self._max_depth = max_depth
    max_verify_length = property(_get_max_verify_length, _set_max_verify_length)
    del _get_max_verify_length, _set_max_verify_length

    def _get_max_verify_bits(self):
        return self._max_bits
    @method_args(int)
    def _set_max_verify_bits(self, max_bits):
        gnutls_certificate_set_verify_limits(self._c_object, max_bits, self._max_depth)
        self._max_bits = max_bits
    max_verify_bits = property(_get_max_verify_bits, _set_max_verify_bits)
    del _get_max_verify_bits, _set_max_verify_bits

    # Methods to select and validate certificates

    def check_certificate(self, cert, cert_name='certificate'):
        """Verify activation, expiration and revocation for the given certificate"""
        now = time()
        if cert.activation_time > now:
            raise CertificateExpiredError("%s is not yet activated" % cert_name)
        if cert.expiration_time < now:
            raise CertificateExpiredError("%s has expired" % cert_name)
        for crl in self.crl_list:
            crl.check_revocation(cert, cert_name=cert_name)

    def select_server_identity(self, session):
        """Select which identity the server will use for a given session. The default selection algorithm uses
        the server name extension. A subclass can overwrite it if a different selection algorithm is desired."""
        server_name = session.server_name
        if server_name is not None:
            return self.server_name_identities.get(server_name)
        elif self.cert and self.key:
            return self ## since we have the cert and key attributes we can behave like a X509Identity
        else:
            return None


class SessionParams(object):
    _default_kx_algorithms = {
        CRED_CERTIFICATE: (KX_RSA, KX_DHE_DSS, KX_DHE_RSA),
        CRED_ANON: (KX_ANON_DH,)}
    _all_kx_algorithms = {
        CRED_CERTIFICATE: set((KX_RSA, KX_DHE_DSS, KX_DHE_RSA, KX_RSA_EXPORT)),
        CRED_ANON: set((KX_ANON_DH,))}

    def __new__(cls, credentials_type):
        if credentials_type not in cls._default_kx_algorithms:
            raise TypeError("Unknown credentials type: %r" % credentials_type)
        return object.__new__(cls)

    def __init__(self, credentials_type):
        self._credentials_type = credentials_type
        self._protocols = (PROTO_TLS1_1, PROTO_TLS1_0, PROTO_SSL3)
        self._kx_algorithms = self._default_kx_algorithms[credentials_type]
        self._ciphers = (CIPHER_AES_128_CBC, CIPHER_3DES_CBC, CIPHER_ARCFOUR_128)
        self._mac_algorithms = (MAC_SHA1, MAC_MD5, MAC_RMD160)
        self._compressions = (COMP_NULL,)

    def _get_protocols(self):
        return self._protocols
    def _set_protocols(self, protocols):
        self._protocols = ProtocolListValidator(protocols)
    protocols = property(_get_protocols, _set_protocols)
    del _get_protocols, _set_protocols

    def _get_kx_algorithms(self):
        return self._kx_algorithms
    def _set_kx_algorithms(self, algorithms):
        cred_type = self._credentials_type
        algorithms = KeyExchangeListValidator(algorithms)
        invalid = set(algorithms) - self._all_kx_algorithms[cred_type]
        if invalid:
            raise ValueError("Cannot specify %r with %r credentials" % (list(invalid), cred_type))
        self._kx_algorithms = algorithms
    kx_algorithms = property(_get_kx_algorithms, _set_kx_algorithms)
    del _get_kx_algorithms, _set_kx_algorithms

    def _get_ciphers(self):
        return self._ciphers
    def _set_ciphers(self, ciphers):
        self._ciphers = CipherListValidator(ciphers)
    ciphers = property(_get_ciphers, _set_ciphers)
    del _get_ciphers, _set_ciphers

    def _get_mac_algorithms(self):
        return self._mac_algorithms
    def _set_mac_algorithms(self, algorithms):
        self._mac_algorithms = MACListValidator(algorithms)
    mac_algorithms = property(_get_mac_algorithms, _set_mac_algorithms)
    del _get_mac_algorithms, _set_mac_algorithms

    def _get_compressions(self):
        return self._compressions
    def _set_compressions(self, compressions):
        self._compressions = CompressionListValidator(compressions)
    compressions = property(_get_compressions, _set_compressions)
    del _get_compressions, _set_compressions


class Session(object):
    """Abstract class representing a TLS session created from a TCP socket
       and a Credentials object."""

    session_type = None ## placeholder for GNUTLS_SERVER or GNUTLS_CLIENT as defined by subclass

    def __new__(cls, *args, **kwargs):
        if cls is Session:
            raise RuntimeError("Session cannot be instantiated directly")
        instance = object.__new__(cls)
        instance.__deinit = gnutls_deinit
        instance._c_object = gnutls_session_t()
        return instance

    def __init__(self, socket, credentials):
        gnutls_init(byref(self._c_object), self.session_type)
        ## Store a pointer to self on the C session
        gnutls_session_set_ptr(self._c_object, id(self))
        # gnutls_dh_set_prime_bits(session, DH_BITS)?
        gnutls_transport_set_ptr(self._c_object, socket.fileno())
        gnutls_handshake_set_private_extensions(self._c_object, 1)
        self.socket = socket
        self.credentials = credentials
        self._update_params()

    def __del__(self):
        self.__deinit(self._c_object)

    def __getattr__(self, name):
        ## Generic wrapper for the underlying socket methods and attributes.
        return getattr(self.socket, name)

    # Session properties

    def _get_credentials(self):
        return self._credentials
    @method_args(X509Credentials)
    def _set_credentials(self, credentials):
        ## Release all credentials, otherwise gnutls will only release an existing credential of
        ## the same type as the one being set and we can end up with multiple credentials in C.
        gnutls_credentials_clear(self._c_object)
        gnutls_credentials_set(self._c_object, credentials._type, cast(credentials._c_object, c_void_p))
        self._credentials = credentials
    credentials = property(_get_credentials, _set_credentials)
    del _get_credentials, _set_credentials

    @property
    def protocol(self):
        return gnutls_protocol_get_name(gnutls_protocol_get_version(self._c_object))

    @property
    def kx_algorithm(self):
        return gnutls_kx_get_name(gnutls_kx_get(self._c_object))

    @property
    def cipher(self):
        return gnutls_cipher_get_name(gnutls_cipher_get(self._c_object))

    @property
    def mac_algorithm(self):
        return gnutls_mac_get_name(gnutls_mac_get(self._c_object))

    @property
    def compression(self):
        return gnutls_compression_get_name(gnutls_compression_get(self._c_object))

    @property
    def peer_certificate(self):
        if gnutls_certificate_type_get(self._c_object) != GNUTLS_CRT_X509:
            return None
        list_size = c_uint()
        cert_list = gnutls_certificate_get_peers(self._c_object, byref(list_size))
        if list_size.value == 0:
            return None
        cert = cert_list[0]
        return X509Certificate(string_at(cert.data, cert.size), X509_FMT_DER)

    # Status checking after an operation was interrupted (these properties are
    # only useful to check after an operation was interrupted, otherwise their
    # value is meaningless).

    @property
    def interrupted_while_writing(self):
        """True if an operation was interrupted while writing"""
        return gnutls_record_get_direction(self._c_object)==1

    @property
    def interrupted_while_reading(self):
        """True if an operation was interrupted while reading"""
        return gnutls_record_get_direction(self._c_object)==0

    # Session methods

    def _update_params(self):
        """Update the priorities of the session params using the credentials."""
        def c_priority_list(priorities):
            size = len(priorities) + 1
            return (c_int * size)(*priorities)
        session_params = self.credentials.session_params
        # protocol order in the priority list is irrelevant (it always uses newer protocols first)
        # the protocol list only specifies what protocols are to be enabled.
        gnutls_protocol_set_priority(self._c_object, c_priority_list(session_params.protocols))
        gnutls_kx_set_priority(self._c_object, c_priority_list(session_params.kx_algorithms))
        gnutls_cipher_set_priority(self._c_object, c_priority_list(session_params.ciphers))
        gnutls_mac_set_priority(self._c_object, c_priority_list(session_params.mac_algorithms))
        gnutls_compression_set_priority(self._c_object, c_priority_list(session_params.compressions))

    def handshake(self):
        gnutls_handshake(self._c_object)

    #@method_args((basestring, buffer))
    def send(self, data):
        data = str(data)
        return gnutls_record_send(self._c_object, data, len(data))

    def sendall(self, data):
        size = len(data)
        while size > 0:
            sent = self.send(data[-size:])
            size -= sent

    def recv(self, limit):
        data = create_string_buffer(limit)
        size = gnutls_record_recv(self._c_object, data, limit)
        return data[:size]

    def send_alert(self, exception):
        alertdict = {
            CertificateError: GNUTLS_A_BAD_CERTIFICATE,
            CertificateAuthorityError: GNUTLS_A_UNKNOWN_CA,
            CertificateSecurityError: GNUTLS_A_INSUFFICIENT_SECURITY,
            CertificateExpiredError: GNUTLS_A_CERTIFICATE_EXPIRED,
            CertificateRevokedError: GNUTLS_A_CERTIFICATE_REVOKED}
        alert = alertdict.get(exception.__class__)
        if alert:
            gnutls_alert_send(self._c_object, GNUTLS_AL_FATAL, alert)

    @method_args(one_of(SHUT_RDWR, SHUT_WR))
    def bye(self, how=SHUT_RDWR):
        gnutls_bye(self._c_object, how)

    def shutdown(self, how=SOCKET_SHUT_RDWR):
        self.socket.shutdown(how)

    def close(self):
        self.socket.close()

    def verify_peer(self):
        status = c_uint()
        gnutls_certificate_verify_peers2(self._c_object, byref(status))
        status = status.value
        if status & GNUTLS_CERT_INVALID:
            raise CertificateError("peer certificate is invalid")
        elif status & GNUTLS_CERT_SIGNER_NOT_FOUND:
            raise CertificateAuthorityError("peer certificate signer not found")
        elif status & GNUTLS_CERT_SIGNER_NOT_CA:
            raise CertificateAuthorityError("peer certificate signer is not a CA")
        elif status & GNUTLS_CERT_INSECURE_ALGORITHM:
            raise CertificateSecurityError("peer certificate uses an insecure algorithm")
        elif status & GNUTLS_CERT_REVOKED:
            raise CertificateRevokedError("peer certificate was revoked")


class ClientSession(Session):
    session_type = GNUTLS_CLIENT

    def __init__(self, socket, credentials, server_name=None):
        Session.__init__(self, socket, credentials)
        self._server_name = None
        if server_name is not None:
            self.server_name = server_name

    def _get_server_name(self):
        return self._server_name
    @method_args(str)
    def _set_server_name(self, server_name):
        gnutls_server_name_set(self._c_object, GNUTLS_NAME_DNS, c_char_p(server_name), len(server_name))
        self._server_name = server_name
    server_name = property(_get_server_name, _set_server_name)
    del _get_server_name, _set_server_name


class ServerSession(Session):
    session_type = GNUTLS_SERVER

    def __init__(self, socket, credentials):
        Session.__init__(self, socket, credentials)
        gnutls_certificate_server_set_request(self._c_object, CERT_REQUEST)

    @property
    def server_name(self):
        data_length = c_size_t(256)
        data = create_string_buffer(data_length.value)
        hostname_type = c_uint()
        for i in xrange(2**16):
            try:
                gnutls_server_name_get(self._c_object, data, byref(data_length), byref(hostname_type), i)
            except RequestedDataNotAvailable:
                break
            except MemoryError:
                data_length.value += 1 ## one extra byte for the terminating 0
                data = create_string_buffer(data_length.value)
                gnutls_server_name_get(self._c_object, data, byref(data_length), byref(hostname_type), i)
            if hostname_type.value != GNUTLS_NAME_DNS:
                continue
            return data.value
        return None


class ServerSessionFactory(object):

    def __init__(self, socket, credentials, session_class=ServerSession):
        if not issubclass(session_class, ServerSession):
            raise TypeError, "session_class must be a subclass of ServerSession"
        self.socket = socket
        self.credentials = credentials
        self.session_class = session_class

    def __getattr__(self, name):
        ## Generic wrapper for the underlying socket methods and attributes
        return getattr(self.socket, name)

    def bind(self, address):
        self.socket.bind(address)

    def listen(self, backlog):
        self.socket.listen(backlog)

    def accept(self):
        new_sock, address = self.socket.accept()
        session = self.session_class(new_sock, self.credentials)
        return (session, address)

    def shutdown(self, how=SOCKET_SHUT_RDWR):
        self.socket.shutdown(how)

    def close(self):
        self.socket.close()